package com.atguigu.flink.java.chapter_7;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/12 20:29
 */
public class Flink09_Chapter07_ProcessWindowFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(2);

        env
          .socketTextStream("mac", 9999)
          .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
              @Override
              public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                  Arrays.stream(value.split("\\W+")).forEach(word -> out.collect(Tuple2.of(word, 1L)));
              }
          })
          .keyBy(t -> t.f0)
          .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
          .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {
              // 参数1: key 参数2: 上下文对象 参数3: 这个窗口内所有的元素 参数4: 收集器, 用于向下游传递数据
              @Override
              public void process(String key,
                                  Context context,
                                  Iterable<Tuple2<String, Long>> elements,
                                  Collector<Tuple2<String, Long>> out) throws Exception {
                  System.out.println(context.window().getStart());
                  long sum = 0L;
                  for (Tuple2<String, Long> t : elements) {
                      sum += t.f1;
                  }
                  out.collect(Tuple2.of(key, sum));
              }
          })
          .print();


        env.execute();
    }
}
